package com;

import com.domain.Order;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author 罗俊华
 * @date 2021/12/16 - 2:11 下午
 */
@Slf4j
public class BasicProducer {

    public static final String TOPIC_NAME = "multiple-partition-replication-topic";
    public static final String BROKERS = "192.168.50.200:9090,192.168.50.200:9091,192.168.50.200:9092";

    @Test
    public void producer() throws JsonProcessingException, InterruptedException {


        Properties properties = new Properties();

        /*
         * broker 的地址可以配置任意多个，kafka client 会自动获得其余的 kafka broker 的地址，并往其发送消息
         * 这里配置多个 ip 地址的目的在于
         * 如果其中一个broker 挂了，可以通过其他broker ip 获取 其余的所有 broker 地址，并向其发送消息
         */
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERS);

        properties.put(ProducerConfig.ACKS_CONFIG, "-1");

//        properties.put(ProducerConfig.ync)

        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        properties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 300);


//        设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送的性能
//        默认为33554432，即32MB
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

//        kafka 本地线程会从缓冲区读取数据，批量发送到broker
//        设置线程批量发送消息的大小，默认是16384，即16kb
//        也就是说，缓冲区中要发送的消息满了16kb就一次发送出去
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

//      默认值是0，代表：消息必须立即被发送，但这样会影响性能
//        一般设置10ms左右，就是说，这个消息发送完之后会进入本地的一个batch
//        如果 10ms 以内这个 batch 满了 16kb 就会随 batch 一起被提前发送出去
//        如果 10ms 以内，batch没满，那么也必须把消息发送出去，不能让消息发送的延迟时间过长
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);


        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        ObjectMapper objectMapper = new ObjectMapper();

        int count = 100;
        CountDownLatch countDownLatch = new CountDownLatch(count);

        for (long i = 0; i < count; i++) {


            Order order = new Order(i, "值" + i);

            /**
             * 未指定发送的分区：
             * 具体发送分区的计算公式：hash(key)%partitionNumber
             * key:具体决定了往哪一个partition上发送消息
             */
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, order.getId().toString(), objectMapper.writeValueAsString(order));

            producer.send(record, (RecordMetadata metadata, Exception e) -> {

                log.info("发往了{}的{}分区，偏移量是：{}", metadata.topic(), metadata.partition(), metadata.offset());

                if (e != null) {
                    log.error("异常", e);
                }

                countDownLatch.countDown();
            });

//            TimeUnit.SECONDS.sleep(1);
        }

        countDownLatch.await();
    }
}
